package com.feeyo.prophet.service;

import com.basho.riak.client.api.RiakClient;
import com.basho.riak.client.api.commands.kv.DeleteValue;
import com.basho.riak.client.api.commands.kv.FetchValue;
import com.basho.riak.client.api.commands.kv.ListKeys;
import com.basho.riak.client.api.commands.kv.StoreValue;
import com.basho.riak.client.core.query.Location;
import com.basho.riak.client.core.query.Namespace;
import com.basho.riak.client.core.query.RiakObject;
import com.basho.riak.client.core.util.BinaryValue;
import com.fasterxml.jackson.core.JsonProcessingException;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.util.concurrent.ExecutionException;

@Slf4j
@Service
public class RiakService {

    @Value("${riak.preview.hosts}")
    private String previewHost;

    private final RiakClient riakClient;

    @Autowired
    public RiakService(RiakClient riakClient) {
        this.riakClient = riakClient;
    }

    public void set(String bucket, String key, byte[] value, String contentType) throws ExecutionException, InterruptedException, JsonProcessingException {
        Location location = new Location(new Namespace(bucket), key);
        RiakObject riakObject = new RiakObject();
        riakObject.setValue(BinaryValue.create(value));
        // 设置Content-Type
        if(contentType != null){
            riakObject.setContentType(contentType);
        }
        // 使用乐观锁
        FetchValue fetch = new FetchValue.Builder(location).build();
        FetchValue.Response fetchResponse = riakClient.execute(fetch);
        RiakObject existingObject = fetchResponse.getValue(RiakObject.class);
        // 设置版本控制
        if (existingObject != null) {
            riakObject.setVClock(existingObject.getVClock());
        }
        StoreValue sv = new StoreValue.Builder(riakObject).withLocation(location).build();
        StoreValue.Response svResponse = this.riakClient.execute(sv);
        log.info("set result: {}", svResponse);
    }

    public RiakObject get(String bucket, String key) throws ExecutionException, InterruptedException {
        Location location = new Location(new Namespace(bucket), key);
        FetchValue fv = new FetchValue.Builder(location).build();
        return this.riakClient.execute(fv).getValue(RiakObject.class);
    }

    /**
     * 预览
     * https://file.veryzhun.com/buckets/prophet/keys/1720771988518.html
     */
    public String preview(String bucket, String key) {
        return String.format("%s/buckets/%s/keys/%s", previewHost, bucket, key);
    }

    /**
     * 删除过期
     * @param bucket
     * @param lastTime
     */
    public void deleteExpiration(String bucket, Long lastTime) throws ExecutionException, InterruptedException {
        Namespace namespace = new Namespace(bucket);
        // 列出 bucket 中的所有 keys
        ListKeys listKeys = new ListKeys.Builder(namespace).build();
        ListKeys.Response keysResponse = riakClient.execute(listKeys);
        for (Location location : keysResponse) {
            BinaryValue key = location.getKey();
            FetchValue fetchValue = new FetchValue.Builder(location).build();
            FetchValue.Response response = riakClient.execute(fetchValue);
            if (response.isNotFound()) {
                continue;
            }
            RiakObject riakObject = response.getValue(RiakObject.class);
            long lastModified = riakObject.getLastModified();
            if(System.currentTimeMillis() - lastModified > lastTime){
                remove(bucket, key.toString());
            }
        }
    }

    public void remove(String bucket, String key) throws ExecutionException, InterruptedException {
        Location location = new Location(new Namespace(bucket), key);
        DeleteValue dv = new DeleteValue.Builder(location).build();
        this.riakClient.execute(dv);
    }
}